package com.rrd.dw.mr.gzip;

import java.io.*;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;

import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.*;
import org.apache.hadoop.hbase.util.*;
import org.apache.hadoop.hbase.mapreduce.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;

import com.alibaba.fastjson.*;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.MD5Util;

public class RongReportsExtractor extends Configured implements Tool {
    private static final String STR_FILED_SPLIT_FLAG = "\001";
    private static final String STR_TB_PREFIX_FLAG = "\007";
    private static final String STR_TARGET_POSITION_BASIC_INF = "basicinfo";
    private static final String STR_TARGET_POSITION_USER_PORTRAIT = "userportrait";
    private static final String STR_TARGET_POSITION_RISK_ANALYSIS = "riskanalysis";
    private static final String STR_TARGET_POSITION_EMERGENCY_ANALYSIS = "emergencyanalysis";
    private static final String STR_TARGET_POSITION_MONTHLY_CONSUMPTION = "monthlyconsumption";
    private static final String STR_TARGET_POSITION_AREA_ANALYSIS = "areaanalysis";
    private static final String STR_TARGET_POSITION_JOB_INF = "jobinfo";
    private static final String STR_TARGET_POSITION_SPECIAL_CALL_INF = "specialcallinfo";

    private static final byte[] FAMILY_INF = Bytes.toBytes("inf");
    private static final byte[] FAMILY_DEC = Bytes.toBytes("dec");
    private static final byte[] FAMILY_JOB = Bytes.toBytes("job");
    private static final byte[] COLUMN_DESC_USER_PORTRAIT = Bytes
        .toBytes("user_portrait");
    private static final byte[] COLUMN_DESC_RISK_ANALYSIS = Bytes
        .toBytes("risk_analysis");
    private static final byte[] COLUMN_DESC_EMERGENCY_ANALYSIS = Bytes
        .toBytes("emergency_analysis");
    private static final byte[] COLUMN_DESC_MONTHLY_CONSUMPTION = Bytes
        .toBytes("monthly_consumption");
    private static final byte[] COLUMN_DESC_AREA_ANALYSIS = Bytes
        .toBytes("area_nalysis");

    private static class RongReportsMapper extends TableMapper<Text, Text> {
        private Text key = null;
        private Text value = null;

        protected void setup(Context context) {
            key = new Text();
            value = new Text();
        }

        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {
            String keyStr = null;
            StringBuffer keysb = new StringBuffer();

            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("i_id_card")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_id_card"))));
            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("i_user_name")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_user_name"))));
            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("i_phone")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("i_phone"))));
            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("h_user_type")) == null
                    ? ""
                    : String.valueOf(Bytes.toInt(value.getValue(FAMILY_INF,
                        Bytes.toBytes("h_user_type")))));
            keysb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("b_reg_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_reg_time")))));
            keyStr = MD5Util.getMd5(keysb.toString());
            this.key.set(keyStr);

            StringBuffer sb = new StringBuffer();
            sb.append(STR_TARGET_POSITION_BASIC_INF + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("i_id_card")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("i_id_card"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("i_user_name")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_user_name"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("i_phone")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("i_phone"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("h_user_type")) == null
                    ? ""
                    : String.valueOf(Bytes.toInt(value.getValue(FAMILY_INF,
                        Bytes.toBytes("h_user_type")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("b_reg_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_reg_time")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("i_emergency_relation2")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_emergency_relation2"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("i_emergency_relation1")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_emergency_relation1"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("i_emergency_phone2")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_emergency_phone2"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("i_emergency_phone1")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_emergency_phone1"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("i_emergency_name2")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_emergency_name2"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("i_emergency_name1")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("i_emergency_name1"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("h_search_id")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("h_search_id"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("h_report_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_INF,
                        Bytes.toBytes("h_report_time")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("b_real_name")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_real_name"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_phone_location")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_phone_location"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("b_phone")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("b_phone"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_operator_zh")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_operator_zh"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("b_operator")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_operator"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_name_check")) == null
                    ? ""
                    : String.valueOf(Bytes.toInt(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_name_check")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_if_call_emergency2")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_if_call_emergency2"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_if_call_emergency1")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_if_call_emergency1"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_id_card_check")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_id_card_check"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_INF, Bytes.toBytes("b_id_card")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_INF, Bytes.toBytes("b_id_card"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_current_balance")) == null
                    ? ""
                    : String.valueOf(Bytes.toFloat(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_current_balance")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(value.getValue(FAMILY_INF,
                Bytes.toBytes("b_ave_monthly_consumption")) == null ? ""
                    : String.valueOf(Bytes.toFloat(value.getValue(FAMILY_INF,
                        Bytes.toBytes("b_ave_monthly_consumption")))));

            this.value.set(sb.toString());

            context.write(this.key, this.value);
            parseUserPortrait(context, keyStr,
                value.getValue(FAMILY_DEC, COLUMN_DESC_USER_PORTRAIT));
            parseRiskAnalysis(context, keyStr,
                value.getValue(FAMILY_DEC, COLUMN_DESC_RISK_ANALYSIS));
            parseEmergencyAnalysis(context, keyStr,
                value.getValue(FAMILY_DEC, COLUMN_DESC_EMERGENCY_ANALYSIS));
            parseMonthlyConsumption(context, keyStr,
                value.getValue(FAMILY_DEC, COLUMN_DESC_MONTHLY_CONSUMPTION));
            parseAreaAnalysis(context, keyStr,
                value.getValue(FAMILY_DEC, COLUMN_DESC_AREA_ANALYSIS));
            parseJobInfo(context, keyStr, value);
        }

        private void parseUserPortrait(Context context, String keyStr,
                byte[] userPortraitBytes)
                throws IOException, InterruptedException {
            if (userPortraitBytes == null)
                return;

            JSONObject userPortraitJson = JSON
                .parseObject(Bytes.toString(userPortraitBytes));
            StringBuffer sb = new StringBuffer();
            sb.append(STR_TARGET_POSITION_USER_PORTRAIT + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
            sb.append(userPortraitJson.getString("night_activity_ratio"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(userPortraitJson.getString("both_call_cnt"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(userPortraitJson.getString("night_msg_ratio"));

            this.key.set(keyStr);
            this.value.set(sb.toString());

            context.write(this.key, this.value);

            parseSpecialCallInfo(context, keyStr,
                userPortraitJson.getJSONArray("special_call_info"));
        }

        private void parseRiskAnalysis(Context context, String keyStr,
                byte[] riskAnalysis) throws IOException, InterruptedException {
            if (riskAnalysis == null)
                return;

            JSONObject riskAnalysisJson = JSON
                .parseObject(Bytes.toString(riskAnalysis));
            StringBuffer sb = new StringBuffer();
            sb.append(STR_TARGET_POSITION_RISK_ANALYSIS + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
            sb.append(riskAnalysisJson.getString("blacklist_cnt"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(riskAnalysisJson.getString("loan_record_cnt"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(riskAnalysisJson.getString("searched_cnt"));

            this.key.set(keyStr);
            this.value.set(sb.toString());

            context.write(this.key, this.value);
        }

        private void parseEmergencyAnalysis(Context context, String keyStr,
                byte[] emergencyAnalysis)
                throws IOException, InterruptedException {
            if (emergencyAnalysis == null)
                return;

            StringBuffer sb = new StringBuffer();
            this.key.set(keyStr);

            JSONObject emergencyAnalysisJson = JSON
                .parseObject("{\"emergency_analysis\":"
                    + Bytes.toString(emergencyAnalysis) + "}");
            JSONArray emergencies = emergencyAnalysisJson
                .getJSONArray("emergency_analysis");
            JSONObject emergencyJson = null;
            for (int i = 0; i < emergencies.size(); i++) {
                sb.delete(0, sb.length());
                emergencyJson = emergencies.getJSONObject(i);

                sb.append(STR_TARGET_POSITION_EMERGENCY_ANALYSIS
                    + STR_TB_PREFIX_FLAG);
                sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                sb.append(i + 1).append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("called_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("talk_seconds"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("name"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("talk_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("phone"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("receive_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("called_seconds"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("msg_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("call_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("unknown_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("send_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("last_contact_date"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("call_seconds"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(emergencyJson.getString("first_contact_date"));

                this.value.set(sb.toString());
                context.write(this.key, this.value);
            }
        }

        private void parseMonthlyConsumption(Context context, String keyStr,
                byte[] monthlyConsumption)
                throws IOException, InterruptedException {
            if (monthlyConsumption == null)
                return;

            StringBuffer sb = new StringBuffer();
            this.key.set(keyStr);

            JSONObject monthlyConsumptionJson = JSON
                .parseObject("{\"monthly_consumption\":"
                    + Bytes.toString(monthlyConsumption) + "}");
            JSONArray consumptions = monthlyConsumptionJson
                .getJSONArray("monthly_consumption");
            JSONObject consumptionJson = null;
            for (int i = 0; i < consumptions.size(); i++) {
                sb.delete(0, sb.length());
                consumptionJson = consumptions.getJSONObject(i);

                sb.append(STR_TARGET_POSITION_MONTHLY_CONSUMPTION
                    + STR_TB_PREFIX_FLAG);
                sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("called_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("talk_seconds"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("talk_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("msg_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("called_seconds"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("receive_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("month"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("call_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("unknown_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("call_consumption"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("send_cnt"))
                    .append(STR_FILED_SPLIT_FLAG);
                sb.append(consumptionJson.getString("call_seconds"));

                this.value.set(sb.toString());
                context.write(this.key, this.value);
            }
        }

        private void parseAreaAnalysis(Context context, String keyStr,
                byte[] areaAnalysis) throws IOException, InterruptedException {
            if (areaAnalysis == null)
                return;

            StringBuffer sb = new StringBuffer();
            this.key.set(keyStr);

            JSONObject areaAnalysisJson = JSON.parseObject(
                "{\"area_analysis\":" + Bytes.toString(areaAnalysis) + "}");
            JSONArray areas = areaAnalysisJson.getJSONArray("area_analysis");
            JSONObject areaJson = null;
            JSONArray areaArray = null;
            JSONObject area = null;
            String areaName = null;
            for (int i = 0; i < areas.size(); i++) {
                areaJson = areas.getJSONObject(i);
                areaName = areaJson.getString("area");
                areaArray = areaJson.getJSONArray("detail");
                if (areaArray != null) {
                    for (int j = 0; j < areaArray.size(); j++) {
                        sb.delete(0, sb.length());
                        area = areaArray.getJSONObject(j);

                        sb.append(STR_TARGET_POSITION_AREA_ANALYSIS
                            + STR_TB_PREFIX_FLAG);
                        sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                        sb.append(areaName).append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("month"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("called_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("talk_seconds"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("talk_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("msg_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("called_seconds"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("receive_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("call_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("unknown_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("contact_phone_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("send_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(area.getString("call_seconds"));

                        this.value.set(sb.toString());
                        context.write(this.key, this.value);
                    }
                }
            }
        }

        private void parseJobInfo(Context context, String keyStr, Result value)
                throws IOException, InterruptedException {
            StringBuffer sb = new StringBuffer();
            sb.append(STR_TARGET_POSITION_JOB_INF + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);

            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("job_id"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("type")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("type"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("try_times")) == null
                    ? ""
                    : String.valueOf(Bytes.toInt(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("try_times")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("timestamp")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("timestamp")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("system_id")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("system_id"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("successful")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("successful")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("start_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("start_time")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("requested")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("requested"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("query")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("query"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("message")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("message"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("finished")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("finished")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("end_time")))));

            this.key.set(keyStr);
            this.value.set(sb.toString());

            context.write(this.key, this.value);
        }

        private void parseSpecialCallInfo(Context context, String keyStr,
                JSONArray specialCalls)
                throws IOException, InterruptedException {
            StringBuffer sb = new StringBuffer();
            JSONObject specialCall = null;
            JSONObject details = null;
            Set<Entry<String, Object>> detailEntrySet = null;
            JSONObject obj = null;
            if (specialCalls != null) {
                for (int i = 0; i < specialCalls.size(); i++) {
                    specialCall = specialCalls.getJSONObject(i);
                    details = specialCall.getJSONObject("detail");
                    detailEntrySet = details.entrySet();
                    for (Entry<String, Object> entry : detailEntrySet) {
                        sb.delete(0, sb.length());

                        sb.append(STR_TARGET_POSITION_SPECIAL_CALL_INF
                            + STR_TB_PREFIX_FLAG);
                        sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
                        sb.append(i).append(STR_FILED_SPLIT_FLAG);
                        sb.append(entry.getKey()).append(STR_FILED_SPLIT_FLAG);
                        obj = (JSONObject) entry.getValue();
                        sb.append(obj.getString("called_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("talk_seconds"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("talk_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("msg_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("called_seconds"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("receive_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("call_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("unknown_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("send_cnt"))
                            .append(STR_FILED_SPLIT_FLAG);
                        sb.append(obj.getString("call_seconds"));

                        this.key.set(keyStr);
                        this.value.set(sb.toString());

                        context.write(this.key, this.value);
                    }
                }
            }
        }
    }

    private static class RongReportsReducer
            extends Reducer<Text, Text, Text, Text> {
        private NullWritable key2 = NullWritable.get();
        private MultipleOutputs<Text, Text> mos = null;

        protected void setup(Context context)
                throws IOException, InterruptedException {
            mos = new MultipleOutputs<Text, Text>(context);
        }

        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            mos.close();
        }

        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text value : values) {
                String[] contents = value.toString().split(STR_TB_PREFIX_FLAG);
                if (contents == null)
                    continue;

                if (STR_TARGET_POSITION_BASIC_INF.equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_BASIC_INF, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_USER_PORTRAIT
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_USER_PORTRAIT, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_RISK_ANALYSIS
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_RISK_ANALYSIS, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_EMERGENCY_ANALYSIS
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_EMERGENCY_ANALYSIS, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_MONTHLY_CONSUMPTION
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_MONTHLY_CONSUMPTION,
                        key2, new Text(contents[1]));
                } else if (STR_TARGET_POSITION_AREA_ANALYSIS
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_AREA_ANALYSIS, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_JOB_INF.equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_JOB_INF, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_SPECIAL_CALL_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_SPECIAL_CALL_INF, key2,
                        new Text(contents[1]));
                }
            }
        }
    }

    public int run(String[] args) throws Exception {
        Properties prop = new Properties();
        try (InputStream is = this.getClass().getClassLoader()
            .getResourceAsStream("hbase.properties");) {
            prop.load(is);
        }
        Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
        Configuration conf = job.getConfiguration();
        conf.set("hbase.zookeeper.quorum", "datanode1,datanode2,datanode3");
        conf.set("mapreduce.map.memory.mb", "3072");
        conf.set("mapreduce.reduce.memory.mb", "5120");
        conf.set("hbase.client.scanner.timeout.period", "120");
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs",
            "org.apache.hadoop.io.compress.GzipCodec");

        /*
         * conf.set("param.out.path",
         * "hdfs://10.10.15.10:8020/tmp/output_xxh_rongreport/");
         * conf.set("param.start_date", "2017-09-19");
         * conf.set("param.stop_date", "2017-09-20");
         */
        conf.set("param.out.path", args[0]);
        conf.set("param.start_date", args[1]);
        conf.set("param.stop_date", args[2]);

        String tableName = "rong_reports";
        String outputDir = conf.get("param.out.path");
        System.out.println("开始日期:" + conf.get("param.start_date"));
        System.out.println("结束日期:" + conf.get("param.stop_date"));
        System.out.println("输入HBase表名:" + tableName);
        System.out.println("输出路径:" + outputDir);

        Long startTime = DateUtil
            .formateToTimestamp(conf.get("param.start_date"));
        Long endTime = DateUtil.formateToTimestamp(conf.get("param.stop_date"));

        Path outpath = new Path(outputDir);
        FileSystem fs = outpath.getFileSystem(conf);
        if (fs.exists(outpath)) {
            fs.delete(outpath, true);
        }

        job.setJarByClass(RongReportsExtractor.class);
        Scan scan = new Scan();

        scan.addFamily(FAMILY_INF);
        scan.addColumn(FAMILY_DEC, COLUMN_DESC_USER_PORTRAIT);
        scan.addColumn(FAMILY_DEC, COLUMN_DESC_RISK_ANALYSIS);
        scan.addColumn(FAMILY_DEC, COLUMN_DESC_EMERGENCY_ANALYSIS);
        scan.addColumn(FAMILY_DEC, COLUMN_DESC_MONTHLY_CONSUMPTION);
        scan.addColumn(FAMILY_DEC, COLUMN_DESC_AREA_ANALYSIS);
        scan.addFamily(FAMILY_JOB);

//		scan.setCacheBlocks(false);
//		scan.setCaching(200);
//		scan.setBatch(100);
        scan.setMaxVersions(1);
//		scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(tableName));
        scan.setTimeRange(startTime, endTime + 14400000);

        TableMapReduceUtil.initTableMapperJob(tableName, scan,
            RongReportsMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, outpath);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_BASIC_INF,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_USER_PORTRAIT,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_RISK_ANALYSIS,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_EMERGENCY_ANALYSIS, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_MONTHLY_CONSUMPTION, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_AREA_ANALYSIS,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_JOB_INF,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_SPECIAL_CALL_INF, TextOutputFormat.class,
            Text.class, Text.class);

        job.setReducerClass(RongReportsReducer.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String... args) throws Exception {
        System.exit(ToolRunner.run(HBaseConfiguration.create(),
            new RongReportsExtractor(), args));
    }
}
